﻿using JESAI.DistributedLock.Internal;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace JESAI.DistributedLock.ZooKeeper;

/// <summary>
/// 基于ZooKeeper的<see cref="IDistributedSemaphore"/>实现。使用与<see cref="ZooKeeperDistributedLock"/>类似的方法。
/// </summary>
public sealed partial class ZooKeeperDistributedSemaphore : IInternalDistributedSemaphore<ZooKeeperDistributedSemaphoreHandle>
{
    private readonly ZooKeeperSynchronizationHelper _synchronizationHelper;

    /// <summary>
    /// 根据提供的<paramref name="path"/>, <paramref name="connectionString"/>和<paramref name="options"/>构造一个新的信号量。
    /// 
    /// 如果指定了<paramref name="assumePathExists"/>, 则在获取锁时不会创建节点，释放锁时也不会删除该节点（默认为false）。
    /// </summary>
    public ZooKeeperDistributedSemaphore(
        ZooKeeperPath path,
        int maxCount,
        string connectionString,
        bool assumePathExists = false,
        Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? options = null)
        : this(path, maxCount, assumePathExists: assumePathExists, connectionString, options)
    {
        if (path == default) { throw new ArgumentNullException(nameof(path)); }
        if (path == ZooKeeperPath.Root) { throw new ArgumentException("不能是根节点", nameof(path)); }
    }

    /// <summary>
    /// 根据提供的<paramref name="name"/>, <paramref name="connectionString"/>和<paramref name="options"/>构造一个新的信号量。
    /// 
    /// 信号量的路径将是根目录'/'下的一个父节点。如果<paramref name="name"/>不是一个有效的节点名称，它将被转换以确保其有效性。
    /// </summary>
    public ZooKeeperDistributedSemaphore(string name, int maxCount, string connectionString, Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? options = null)
        : this(ZooKeeperPath.Root, name, maxCount, connectionString, options)
    {
    }

    /// <summary>
    /// 根据提供的<paramref name="directoryPath"/>, <paramref name="name"/>, <paramref name="connectionString"/>和<paramref name="options"/>构造一个新的信号量。
    /// 
    /// 信号量的路径将是<paramref name="directoryPath"/>下的一个父节点。如果<paramref name="name"/>不是一个有效的节点名称，它将被转换以确保其有效性。
    /// </summary>
    public ZooKeeperDistributedSemaphore(ZooKeeperPath directoryPath, string name, int maxCount, string connectionString, Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? options = null)
        : this(
              (directoryPath == default ? throw new ArgumentNullException(nameof(directoryPath)) : directoryPath).GetChildNodePathWithSafeName(name),
              maxCount,
              assumePathExists: false,
              connectionString,
              options)
    {
    }

    private ZooKeeperDistributedSemaphore(ZooKeeperPath nodePath, int maxCount, bool assumePathExists, string connectionString, Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? optionsBuilder)
    {
        if (maxCount < 1) { throw new ArgumentOutOfRangeException(nameof(maxCount), maxCount, "必须为正数"); }
        this.MaxCount = maxCount;
        // setAcquiredMarker是必需的，因为我们使用数据变化作为等待过程的一部分
        this._synchronizationHelper = new ZooKeeperSynchronizationHelper(nodePath, assumePathExists, connectionString, optionsBuilder, setAcquiredMarker: true);
    }

    /// <summary>
    /// ZooKeeper节点路径
    /// </summary>
    public ZooKeeperPath Path => this._synchronizationHelper.Path;

    /// <summary>
    /// 实现<see cref="IDistributedSemaphore.Name"/>。显式实现以避免与传递的名称包括前导"/"和基础目录的混淆。
    /// </summary>
    string IDistributedSemaphore.Name => this.Path.ToString();

    /// <summary>
    /// 实现<see cref="IDistributedSemaphore.MaxCount"/>
    /// </summary>
    public int MaxCount { get; }

    async ValueTask<ZooKeeperDistributedSemaphoreHandle?> IInternalDistributedSemaphore<ZooKeeperDistributedSemaphoreHandle>.InternalTryAcquireAsync(TimeoutValue timeout, CancellationToken cancellationToken)
    {
        var nodeHandle = await this._synchronizationHelper.TryAcquireAsync(
                hasAcquired: state => Array.FindIndex(state.SortedChildren, t => t.Path == state.EphemeralNodePath) < this.MaxCount,
                waitAsync: async (zooKeeper, state, watcher) =>
                {
                    var ephemeralNodeIndex = Array.FindIndex(state.SortedChildren, t => t.Path == state.EphemeralNodePath);
                    Invariant.Require(ephemeralNodeIndex >= this.MaxCount);

                    // 如果我们是下一个等待节点，则等待任何子节点集合的变化
                    if (ephemeralNodeIndex == this.MaxCount)
                    {
                        var childNames = new HashSet<string>((await zooKeeper.getChildrenAsync(this.Path.ToString(), watcher).ConfigureAwait(false)).Children);
                        // 如果我们前面的任何子节点缺失，则等待完成。否则，
                        // 让监视器在子节点集合发生变化时通知我们
                        return state.SortedChildren.Take(ephemeralNodeIndex)
                            .Any(t => !childNames.Contains(t.Path.Substring(t.Path.LastIndexOf(ZooKeeperPath.Separator) + 1)));
                    }

                    // 否则，我们只监视我们前面的节点的数据变化。虽然我们可以
                    // 在这种情况下监视所有子节点，但这种方法效率较低，因为它会导致
                    // 群体效应，每个新的等待者或释放等待者都会唤醒其他人
                    var nextLowestChildData = await zooKeeper.getDataAsync(state.SortedChildren[ephemeralNodeIndex - 1].Path, watcher).ConfigureAwait(false);
                    // 如果它已经获取了锁，则等待完成。否则，监视器将在数据变化或删除该节点时通知我们
                    return nextLowestChildData.Data.SequenceEqual(ZooKeeperSynchronizationHelper.AcquiredMarker);
                },
                timeout,
                cancellationToken,
                nodePrefix: "semaphore-"
            )
            // 我们被迫在这里使用async-over-sync，因为ZooKeeperNetEx没有同步API
            .AwaitSyncOverAsync()
            .ConfigureAwait(false);

        return nodeHandle != null ? new ZooKeeperDistributedSemaphoreHandle(nodeHandle) : null;
    }
}
